package com.ustcinfo.study.hbase.r1.zhangshouxin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;

import java.io.IOException;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
/**
 * Describe:HBase存数据与批量存入
 * Author: 张守信
 */
public class HbasePutInfo {
    /*集群配置对象*/
    private static Configuration conf = null;

    /*集群连接对象*/
    private static Connection connection = null;

    /*hbase集群管理对象*/
    private static Admin admin = null;

    /*hbase replication管理对象*/
    private static ReplicationAdmin replicationAdmin = null;

    /**
     * 初始化
     */
    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "nodex,nodex1,nodex2");
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            replicationAdmin = new ReplicationAdmin(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /*
    连接本机的mysql,建立连接读取表数据
    */
    private static java.sql.Connection connectDB() throws Exception {
        String userName="root";
        String password="123456";
        String url="jdbc:mysql://127.0.0.1:3306/test";
        Class.forName("com.mysql.jdbc.Driver").newInstance();
        java.sql.Connection conn=DriverManager.getConnection(url,userName,password);
        System.out.println(conn);
        return conn;
    }
    /*
    讲为null的数据变为空字符串
    */
    public static  String transfer(String x){
        if(x==null)
        {
            x="";
        }
        return x;
    }
//    hbase批量插入数据
    public static void putList(String name, List<Put> puts) {
        TableName tableName = TableName.valueOf(name);
        try (Table table = connection.getTable(tableName)) {
            for (Put put : puts) {
                table.put(put);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /*
    hbase插入单条数据
    */
    public static void put(String name, Put put) {
        TableName tableName = TableName.valueOf(name);
        try (Table table = connection.getTable(tableName)) {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws SQLException {

        java.sql.Connection conn=null;
        Statement statement=null;
        try {
            conn=connectDB();
        } catch (Exception e) {
            e.printStackTrace();
        }

        String query="select * from transfer_task_to";//查询transfer_task_to表数据
        try {
            statement= conn.createStatement();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        ResultSet rs= null;//sql查询的结果集
        try {
            rs = statement.executeQuery(query);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        List<Put> puts=new ArrayList<>();//存放批量的put


        while (rs.next()){
            String id=rs.getString("id");//获取id值

            Put put=new Put(id.getBytes());//以id为rowkey
            put.addColumn("F1".getBytes(),"id".getBytes(), transfer(rs.getString(1)).getBytes());
            put.addColumn("F1".getBytes(),"task_id".getBytes(), transfer(rs.getString(3)).getBytes());
            put.addColumn("F1".getBytes(),"create_time".getBytes(), transfer(rs.getString(4)).getBytes());
            put.addColumn("F1".getBytes(),"start_time".getBytes(), transfer( rs.getString(11)).getBytes());
            put.addColumn("F1".getBytes(),"out_filter_number".getBytes(), rs.getString(13).getBytes());
            put.addColumn("F1".getBytes(),"from_run_param".getBytes(), rs.getString(14).getBytes());

            put.addColumn("F2".getBytes(),"current_retry_times".getBytes(), rs.getString(15).getBytes());
            put.addColumn("F2".getBytes(),"retry_flag".getBytes(), rs.getString(17).getBytes());
            put.addColumn("F2".getBytes(),"parent_task_id".getBytes(), rs.getString(18).getBytes());
            put.addColumn("F2".getBytes(),"end_time".getBytes(), transfer(rs.getString(19)).getBytes());
            put.addColumn("F2".getBytes(),"kill_time".getBytes(),transfer( rs.getString(20)).getBytes());
            put.addColumn("F2".getBytes(),"failure_time".getBytes(), transfer(rs.getString(21)).getBytes());
            puts.add(put);
        }

        putList("BD_TEST", puts);//将puts存入list

    }
}
